package com.insight.service;

import com.insight.common.enums.CollectionType;
import com.insight.common.enums.TaskStatus;
import com.insight.datasource.adapter.MetadataExtractor;
import com.insight.entity.CollectionTask;
import com.insight.entity.DataSource;
import com.insight.entity.MetadataColumn;
import com.insight.entity.MetadataIndex;
import com.insight.entity.MetadataTable;
import com.insight.repository.MetadataColumnRepository;
import com.insight.repository.MetadataIndexRepository;
import com.insight.repository.MetadataTableRepository;
import com.insight.dto.collection.FailedTableInfo;
import com.insight.dto.collection.TableDetails;
import com.alibaba.fastjson2.JSON;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import jakarta.persistence.EntityManager;

import java.util.ArrayList;
import java.util.List;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.security.MessageDigest;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;

/**
 * T070: 元数据采集服务实现
 * 负责全量和增量元数据采集,使用Java 21 Virtual Threads进行高并发处理
 *
 * @author Insight Team
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class MetadataCollectionService {

    private final MetadataTableRepository metadataTableRepository;
    private final MetadataColumnRepository metadataColumnRepository;
    private final MetadataIndexRepository metadataIndexRepository;
    private final CollectionTaskService collectionTaskService;

    /**
     * Virtual Threads执行器 - 用于高并发I/O密集型任务
     */
    private final ExecutorService virtualThreadExecutor = Executors.newVirtualThreadPerTaskExecutor();

    /**
     * 全量采集元数据 (使用已存在的任务)
     * 从指定数据源采集所有表的元数据信息,使用已创建的任务记录进度
     *
     * @param taskId 已创建的任务ID
     * @param dataSource 数据源
     * @param extractor  元数据提取器
     * @return 采集任务
     */
    @Transactional
    public CollectionTask collectFullMetadata(Long taskId, DataSource dataSource, MetadataExtractor extractor) {
        // 参数校验
        validateParameters(dataSource, extractor);
        if (taskId == null) {
            throw new IllegalArgumentException("任务ID不能为空");
        }

        log.info("开始全量采集元数据: taskId={}, dataSource={}", taskId, dataSource.getName());

        // 1. 获取并启动已存在的任务
        CollectionTask task = collectionTaskService.getTaskById(taskId)
                .orElseThrow(() -> new IllegalArgumentException("任务不存在: taskId=" + taskId));
        task = collectionTaskService.startTask(taskId);

        try {
            // 2. 提取表列表
            List<MetadataTable> tables = extractor.extractTables(dataSource, null);
            log.info("全量采集发现{}张表: dataSource={}", tables.size(), dataSource.getName());

            // 3. 更新任务总表数
            task = collectionTaskService.updateProgress(taskId, tables.size(), 0);

            // 4. 处理空表情况
            if (tables.isEmpty()) {
                log.warn("数据源中未发现任何表,采集完成: dataSource={}", dataSource.getName());
                task = collectionTaskService.completeTask(taskId);
                return task;
            }

            // 5. 保存表元数据
            List<MetadataTable> savedTables = metadataTableRepository.saveAll(tables);
            log.info("已保存{}张表元数据: dataSource={}", savedTables.size(), dataSource.getName());

            // 6. 使用Virtual Threads并发提取每张表的列和索引数据(不保存)
            List<CompletableFuture<TableDetails>> futures = new ArrayList<>();

            for (MetadataTable table : savedTables) {
                CompletableFuture<TableDetails> future = CompletableFuture.supplyAsync(() -> {
                    return extractTableDetails(dataSource, extractor, table);
                }, virtualThreadExecutor);
                futures.add(future);
            }

            // 7. 等待所有表的数据提取完成
            List<TableDetails> allTableDetails = futures.stream()
                    .map(CompletableFuture::join)
                    .collect(Collectors.toList());

            log.info("数据提取完成,共{}张表", allTableDetails.size());

            // 8. 在主线程的同一事务中批量保存所有列和索引
            List<FailedTableInfo> failedTablesList = new ArrayList<>();
            int successCount = saveAllTableDetails(allTableDetails, failedTablesList);

            // 更新进度到任务服务
            collectionTaskService.updateProgress(taskId, tables.size(), successCount);

            log.info("增量采集完成: dataSource={}, 成功采集{}/{}张表, 失败{}张表",
                    dataSource.getName(), successCount, tables.size(), failedTablesList.size());

            // 8. 批量更新表统计信息到数据库
            try {
                metadataTableRepository.saveAll(savedTables);
                log.info("已批量更新{}张表的统计信息", savedTables.size());
            } catch (Exception e) {
                log.warn("批量更新表统计信息失败: {}", e.getMessage());
            }

            // 9. 保存失败表信息并标记任务完成
            if (!failedTablesList.isEmpty()) {
                String failedTablesJson = JSON.toJSONString(failedTablesList);
                collectionTaskService.updateFailedTables(taskId, failedTablesJson);
            }
            
            task = collectionTaskService.completeTask(taskId);
            return task;

        } catch (Exception e) {
            log.error("全量采集失败: dataSource={}, error={}", dataSource.getName(), e.getMessage(), e);
            collectionTaskService.failTask(taskId, e.getMessage());
            throw new RuntimeException("全量采集失败: " + e.getMessage(), e);
        }
    }

    /**
     * 全量采集元数据 (自动创建任务)
     * 从指定数据源采集所有表的元数据信息,内部创建新任务
     *
     * @param dataSource 数据源
     * @param extractor  元数据提取器
     * @return 采集任务
     */
    @Transactional
    public CollectionTask collectFullMetadata(DataSource dataSource, MetadataExtractor extractor) {
        // 参数校验(必须先校验,再访问dataSource属性)
        validateParameters(dataSource, extractor);

        log.info("开始全量采集元数据: dataSource={}", dataSource.getName());

        // 1. 创建并启动采集任务
        CollectionTask task = collectionTaskService.createTask(dataSource, CollectionType.FULL);
        task = collectionTaskService.startTask(task.getId());

        try {
            // 2. 提取表列表
            List<MetadataTable> tables = extractor.extractTables(dataSource, null);
            log.info("全量采集发现{}张表: dataSource={}", tables.size(), dataSource.getName());

            // 3. 更新任务总表数
            task = collectionTaskService.updateProgress(task.getId(), tables.size(), 0);

            // 4. 处理空表情况
            if (tables.isEmpty()) {
                log.warn("数据源中未发现任何表,采集完成: dataSource={}", dataSource.getName());
                task = collectionTaskService.completeTask(task.getId());
                return task;
            }

            // 5. 保存表元数据
            List<MetadataTable> savedTables = metadataTableRepository.saveAll(tables);
            log.info("已保存{}张表元数据: dataSource={}", savedTables.size(), dataSource.getName());

            // 6. 使用Virtual Threads并发提取每张表的列和索引数据(不保存)
            List<CompletableFuture<TableDetails>> futures = new ArrayList<>();

            for (MetadataTable table : savedTables) {
                CompletableFuture<TableDetails> future = CompletableFuture.supplyAsync(() -> {
                    return extractTableDetails(dataSource, extractor, table);
                }, virtualThreadExecutor);

                futures.add(future);
            }

            // 7. 等待所有表的数据提取完成
            List<TableDetails> allTableDetails = futures.stream()
                    .map(CompletableFuture::join)
                    .collect(Collectors.toList());

            log.info("数据提取完成,共{}张表", allTableDetails.size());

            // 8. 在主线程的同一事务中批量保存所有列和索引
            List<FailedTableInfo> failedTablesList = new ArrayList<>();
            int successCount = saveAllTableDetails(allTableDetails, failedTablesList);

            log.info("全量采集完成: dataSource={}, 成功采集{}/{}张表, 失败{}张表",
                    dataSource.getName(), successCount, tables.size(), failedTablesList.size());

            // 8. 批量更新表统计信息到数据库
            try {
                metadataTableRepository.saveAll(savedTables);
                log.info("已批量更新{}张表的统计信息", savedTables.size());
            } catch (Exception e) {
                log.warn("批量更新表统计信息失败: {}", e.getMessage());
                // 统计信息更新失败不影响主流程
            }

            // 9. 保存失败表信息并标记任务完成
            if (!failedTablesList.isEmpty()) {
                String failedTablesJson = JSON.toJSONString(failedTablesList);
                collectionTaskService.updateFailedTables(task.getId(), failedTablesJson);
            }
            
            task = collectionTaskService.completeTask(task.getId());

            return task;

        } catch (Exception e) {
            log.error("全量采集失败: dataSource={}, error={}", dataSource.getName(), e.getMessage(), e);

            // 标记任务失败
            collectionTaskService.failTask(task.getId(), e.getMessage());

            throw new RuntimeException("全量采集失败: " + e.getMessage(), e);
        }
    }

    /**
     * 增量采集元数据
     * 只采集有变化的表(新增/删除/修改)
     *
     * @param dataSource 数据源
     * @param extractor  元数据提取器
     * @return 采集任务
     */
    @Transactional
    public CollectionTask collectIncrementalMetadata(DataSource dataSource, MetadataExtractor extractor) {
        // 参数校验(必须先校验,再访问dataSource属性)
        validateParameters(dataSource, extractor);

        log.info("开始增量采集元数据: dataSource={}", dataSource.getName());

        // 1. 创建并启动采集任务
        CollectionTask task = collectionTaskService.createTask(dataSource, CollectionType.INCREMENTAL);
        task = collectionTaskService.startTask(task.getId());

        try {
            // 2. 获取当前数据源的所有表
            List<MetadataTable> currentTables = extractor.extractTables(dataSource, null);
            log.info("增量采集发现{}张当前表: dataSource={}", currentTables.size(), dataSource.getName());

            // 3. 获取已存储的表
            List<MetadataTable> existingTables = metadataTableRepository.findByDataSource(dataSource);
            log.info("数据库中已存储{}张表: dataSource={}", existingTables.size(), dataSource.getName());

            // 4. 比对差异
            Map<String, MetadataTable> currentTableMap = currentTables.stream()
                    .collect(Collectors.toMap(MetadataTable::getFullName, t -> t));
            Map<String, MetadataTable> existingTableMap = existingTables.stream()
                    .collect(Collectors.toMap(MetadataTable::getFullName, t -> t));

            // 5. 识别新增的表
            List<MetadataTable> newTables = currentTables.stream()
                    .filter(t -> !existingTableMap.containsKey(t.getFullName()))
                    .collect(Collectors.toList());
            log.info("发现{}张新增表: dataSource={}", newTables.size(), dataSource.getName());

            // 6. 识别删除的表
            List<MetadataTable> deletedTables = existingTables.stream()
                    .filter(t -> !currentTableMap.containsKey(t.getFullName()))
                    .collect(Collectors.toList());
            log.info("发现{}张已删除表: dataSource={}", deletedTables.size(), dataSource.getName());

            // 7. 更新任务总表数(新增+删除)
            int totalChanges = newTables.size() + deletedTables.size();
            task = collectionTaskService.updateProgress(task.getId(), totalChanges, 0);

            // 8. 处理新增表
            if (!newTables.isEmpty()) {
                List<MetadataTable> savedNewTables = metadataTableRepository.saveAll(newTables);
                final Long taskId = task.getId();

                // 使用两阶段模式处理新增表
                List<CompletableFuture<TableDetails>> futures = new ArrayList<>();

                for (MetadataTable table : savedNewTables) {
                    CompletableFuture<TableDetails> future = CompletableFuture.supplyAsync(() -> {
                        return extractTableDetails(dataSource, extractor, table);
                    }, virtualThreadExecutor);
                    futures.add(future);
                }

                // 等待所有表的数据提取完成
                List<TableDetails> allTableDetails = futures.stream()
                        .map(CompletableFuture::join)
                        .collect(Collectors.toList());

                log.info("新增表数据提取完成,共{}张表", allTableDetails.size());

                // 在主线程的同一事务中批量保存
                List<FailedTableInfo> failedTablesList = new ArrayList<>();
                int successCount = saveAllTableDetails(allTableDetails, failedTablesList);

                // 更新进度
                collectionTaskService.updateProgress(taskId, totalChanges, successCount);

                log.info("新增表采集完成: 成功{}/{}张", successCount, savedNewTables.size());
            }

            // 9. 处理删除表
            if (!deletedTables.isEmpty()) {
                for (MetadataTable table : deletedTables) {
                    metadataTableRepository.delete(table);
                    log.info("已删除表元数据: {}", table.getFullName());
                }

                // 更新进度
                collectionTaskService.updateProgress(task.getId(), totalChanges, totalChanges);
            }

            log.info("增量采集完成: dataSource={}, 新增{}张表, 删除{}张表",
                    dataSource.getName(), newTables.size(), deletedTables.size());

            // 10. 标记任务完成
            task = collectionTaskService.completeTask(task.getId());

            return task;

        } catch (Exception e) {
            log.error("增量采集失败: dataSource={}, error={}", dataSource.getName(), e.getMessage(), e);

            // 标记任务失败
            collectionTaskService.failTask(task.getId(), e.getMessage());

            throw new RuntimeException("增量采集失败: " + e.getMessage(), e);
        }
    }

    /**
     * 取消采集任务
     *
     * @param taskId 任务ID
     * @return 取消后的任务
     */
    public CollectionTask cancelCollection(Long taskId) {
        log.info("取消采集任务: taskId={}", taskId);
        return collectionTaskService.cancelTask(taskId);
    }

    /**
     * 查询采集进度
     *
     * @param taskId 任务ID
     * @return 任务Optional
     */
    public Optional<CollectionTask> getCollectionProgress(Long taskId) {
        return collectionTaskService.getTaskById(taskId);
    }

    /**
     * 提取单张表的详细信息(列和索引) - 只提取数据,不保存
     * 在Virtual Thread中调用,避免跨Session的实体引用问题
     *
     * @param dataSource 数据源
     * @param extractor  提取器
     * @param table      表
     * @return TableDetails 包含列和索引的详情对象
     */
    private TableDetails extractTableDetails(DataSource dataSource, MetadataExtractor extractor, MetadataTable table) {
        try {
            // 确保table有ID
            if (table.getId() == null) {
                log.error("表ID为null,无法采集详情: table={}", table.getFullName());
                return TableDetails.failure(table, "表未持久化,ID为空");
            }

            // 1. 提取列信息
            List<MetadataColumn> columns = extractor.extractColumns(dataSource, table);
            log.debug("提取到{}个列: table={}", columns.size(), table.getFullName());

            // 2. 提取索引信息
            List<MetadataIndex> indexes = extractor.extractIndexes(dataSource, table);
            log.debug("提取到{}个索引: table={}", indexes.size(), table.getFullName());

            // 3. 更新表统计信息(仅在内存中)
            updateTableStatistics(table, columns);

            return TableDetails.success(table, columns, indexes);

        } catch (Exception e) {
            log.error("提取表详情失败: table={}, error={}", table.getFullName(), e.getMessage(), e);
            return TableDetails.failure(table, e.getMessage());
        }
    }

    /**
     * 批量保存所有表的列和索引
     * 在主线程的同一事务中执行,避免跨Session的实体引用问题
     *
     * @param allTableDetails 所有表的详情列表
     * @param failedTablesList 失败表信息列表(输出参数)
     * @return 成功保存的表数量
     */
    private int saveAllTableDetails(List<TableDetails> allTableDetails, List<FailedTableInfo> failedTablesList) {
        int successCount = 0;

        // 收集所有的列和索引
        List<MetadataColumn> allColumns = new ArrayList<>();
        List<MetadataIndex> allIndexes = new ArrayList<>();

        for (TableDetails details : allTableDetails) {
            if (!details.isSuccess()) {
                // 记录失败的表
                failedTablesList.add(new FailedTableInfo(
                        details.getTable().getTableName(),
                        details.getTable().getTableSchema(),
                        details.getErrorMessage()
                ));
                continue;
            }

            MetadataTable table = details.getTable();
            List<MetadataColumn> columns = details.getColumns();
            List<MetadataIndex> indexes = details.getIndexes();

            // 设置列的table关联
            if (columns != null && !columns.isEmpty()) {
                for (MetadataColumn column : columns) {
                    column.setTable(table);  // 使用同一Session中的table对象
                }
                allColumns.addAll(columns);
            }

            // 设置索引的table关联
            if (indexes != null && !indexes.isEmpty()) {
                for (MetadataIndex index : indexes) {
                    index.setTable(table);  // 使用同一Session中的table对象
                }
                allIndexes.addAll(indexes);
            }

            successCount++;
        }

        // 批量保存所有列
        if (!allColumns.isEmpty()) {
            metadataColumnRepository.saveAll(allColumns);
            log.info("已批量保存{}个列", allColumns.size());
        }

        // 批量保存所有索引
        if (!allIndexes.isEmpty()) {
            metadataIndexRepository.saveAll(allIndexes);
            log.info("已批量保存{}个索引", allIndexes.size());
        }

        return successCount;
    }

    /**
     * 采集单张表的详细信息(列和索引) - 已废弃,保留用于增量采集
     *
     * @param dataSource 数据源
     * @param extractor  提取器
     * @param table      表
     */
    @Deprecated
    private void collectTableDetails(DataSource dataSource, MetadataExtractor extractor, MetadataTable table) {
        try {
            // 确保table有ID(应该在主线程已保存)
            if (table.getId() == null) {
                log.error("表ID为null,无法采集详情: table={}", table.getFullName());
                throw new IllegalStateException("表未持久化,ID为空: " + table.getFullName());
            }

            // 1. 采集列信息
            List<MetadataColumn> columns = extractor.extractColumns(dataSource, table);
            log.debug("提取到{}个列: table={}", columns.size(), table.getFullName());

            // 保存列 - 创建只包含ID的table引用对象
            if (!columns.isEmpty()) {
                // 创建一个只设置了ID的table对象作为引用
                // Hibernate会将其视为managed状态的引用,不会尝试重新加载
                MetadataTable tableReference = new MetadataTable();
                tableReference.setId(table.getId());

                // 使用table引用设置列的关联
                for (MetadataColumn column : columns) {
                    column.setTable(tableReference);
                }
                metadataColumnRepository.saveAll(columns);
                log.debug("已保存{}个列: table={}", columns.size(), table.getFullName());
            }

            // 2. 采集索引信息
            List<MetadataIndex> indexes = extractor.extractIndexes(dataSource, table);
            log.debug("提取到{}个索引: table={}", indexes.size(), table.getFullName());

            // 保存索引 - 同样创建只包含ID的table引用
            if (!indexes.isEmpty()) {
                // 创建一个只设置了ID的table对象作为引用
                MetadataTable tableReference = new MetadataTable();
                tableReference.setId(table.getId());

                // 使用table引用设置索引的关联
                for (MetadataIndex index : indexes) {
                    index.setTable(tableReference);
                }
                metadataIndexRepository.saveAll(indexes);
                log.debug("已保存{}个索引: table={}", indexes.size(), table.getFullName());
            }

            // 3. 更新表统计信息 (仅更新内存中的对象,不保存到数据库)
            // 将在主线程批量更新所有表的统计信息
            updateTableStatistics(table, columns);
            log.debug("已更新表统计信息(内存): table={}", table.getFullName());

        } catch (Exception e) {
            log.error("采集表详情异常: table={}, error={}", table.getFullName(), e.getMessage());
            throw new RuntimeException("采集表详情失败: " + table.getFullName(), e);
        }
    }


    /**
     * 更新表统计信息
     * 
     * @param table 表元数据
     * @param columns 列列表
     */
    private void updateTableStatistics(MetadataTable table, List<MetadataColumn> columns) {
        try {
            // 1. 更新最后采集时间
            table.setLastCollectedAt(LocalDateTime.now());
            
            // 2. 计算表结构哈希(基于列定义)
            if (!columns.isEmpty()) {
                StringBuilder structureBuilder = new StringBuilder();
                for (MetadataColumn column : columns) {
                    structureBuilder.append(column.getColumnName())
                            .append(":")
                            .append(column.getDataType())
                            .append(":")
                            .append(column.getColumnLength())
                            .append(";");
                }
                
                String structureHash = calculateMD5(structureBuilder.toString());
                table.setStructureHash(structureHash);
            }
            
            // 3. 注意: rowCount 和 dataSizeBytes 需要通过数据库查询获取,这里暂不实现
            // 因为需要执行 SELECT COUNT(*) 和查询表大小的SQL,可能耗时较长
            // 可以作为后续优化项,或者在后台异步任务中执行
            
        } catch (Exception e) {
            log.warn("更新表统计信息失败: table={}, error={}", table.getFullName(), e.getMessage());
            // 统计信息更新失败不影响主流程
        }
    }
    
    /**
     * 计算字符串的MD5哈希值
     * 
     * @param input 输入字符串
     * @return MD5哈希值(十六进制字符串)
     */
    private String calculateMD5(String input) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] hashBytes = md.digest(input.getBytes(StandardCharsets.UTF_8));
            
            StringBuilder hexString = new StringBuilder();
            for (byte b : hashBytes) {
                String hex = Integer.toHexString(0xff & b);
                if (hex.length() == 1) {
                    hexString.append('0');
                }
                hexString.append(hex);
            }
            return hexString.toString();
            
        } catch (Exception e) {
            log.error("计算MD5哈希失败: {}", e.getMessage());
            return null;
        }
    }

    /**
     * 参数校验
     *
     * @param dataSource 数据源
     * @param extractor  提取器
     */
    private void validateParameters(DataSource dataSource, MetadataExtractor extractor) {
        if (dataSource == null) {
            throw new IllegalArgumentException("数据源不能为空");
        }
        if (extractor == null) {
            throw new IllegalArgumentException("元数据提取器不能为空");
        }
    }
}
